package com.icloud.cer.rest.util;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.icloud.cer.rest.Constants;
import com.icloud.cer.rest.common.BizException;
import com.icloud.cer.rest.common.MessageCode;
import com.icloud.cer.rest.msg.MessageHolder;
import com.icloud.cer.rest.msg.MsgConstants;
import com.icloud.cer.rest.msg.bean.EncryptRequest;
import com.icloud.cer.rest.msg.bean.HeartbeatRequest;


public class SenderManager {
	private static final Logger log = LoggerFactory.getLogger(SenderManager.class);
	
	/*
	 * 模拟同步消息，因为都是异步调用,用于判断消息接收是否正常，需要有个线程对未返回的消息进行处理 Key是sequence
	 */
	private ConcurrentHashMap<Long, TimeoutMessage> sentMessages = new ConcurrentHashMap<Long, TimeoutMessage>();
	private ConcurrentHashMap<Long, TimeoutMessage> timeOutMessages = new ConcurrentHashMap<Long, TimeoutMessage>();
	
	// 模拟同步消息，响应的消息队列
	private ConcurrentHashMap<Long, MessageHolder> respnMessages = new ConcurrentHashMap<Long, MessageHolder>();
	
	private ConnectorManager connectManager;
	
	public ConnectorManager getConnectManager() {
		return connectManager;
	}

	public void setConnectManager(ConnectorManager connectManager) {
		this.connectManager = connectManager;
	}

	public SenderManager() {
		// 启动扫描线程
		new CheckTimeOutThread().start();
	}
	
	/**
	 * @param message  请求的消息
	 * @param timeout  等待时长(毫秒)，超过这个时间，就不等待了
	 * @param commandid   等待响应的消息类型，有些响应消息会有多种，只能等待一个消息，其他消息需要异步处理
	 * @return 响应的消息
	 */
	public MessageHolder sendEncyptRequestAndWait(MessageHolder message, long timeout, short commandid) {
		if(connectManager.hasClient()) {
			throw new BizException(MessageCode.E10014);
		}
		// 设置消息ID
		long msgID = MessageSequenceGenerator.getNextMessageId();
		message.setId(msgID);
		message.setSequence(msgID);
		EncryptRequest er = (EncryptRequest)message.getValue();
		er.setSequence(msgID);

		log.info("sequence:" + msgID);
		short sendCommand = MsgConstants.SERVER_SENT_ENCRYPT;
		// 放入队列
		TimeoutMessage tom = new TimeoutMessage(message, timeout, sendCommand, commandid);
		sentMessages.put(msgID, tom);
		// 得到空闲的session
		IoSession session = connectManager.getIdleSession();
		// 如果所有的session都在处理任务，则等待1秒，继续请求空闲的session
		long start = System.currentTimeMillis();
		if(session == null) {
			while(true) {
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				session = connectManager.getIdleSession();
				if(session != null) {
					break;
				}
				if(System.currentTimeMillis() - start > Constants.DEFALUT_TIMEOUT) {
					throw new BizException(MessageCode.E10013);
				}
			}
		}
		
		Long sessionId = session.getId();
		connectManager.putTask(sessionId);
		if(sentMessages.get(msgID) != null) {
			session.write(message);
			//防止加密机主动取走重复的认为
			tom.setStatus(1);
			log.info("sessionId:" + session.getId() + ",address:" +  session.getRemoteAddress());
			// 等待消息的返回
			EncryptUtils.doWait(tom);
			connectManager.popTask(sessionId);
			// 从返回队列中获取返回值
			MessageHolder respnMessage = respnMessages.remove(msgID);
			if (respnMessage != null) {
				return respnMessage;
			} else {
				log.error("response time out");
				throw new BizException(MessageCode.E10013);
			}
		}
		// 任务被加密机主动请求走了或者超时
		log.warn("任务被客户端主动请求走");
		MessageHolder respnMessage = respnMessages.get(msgID);
		//从response队列中取返回结果，如果没有，则继续判断超时队列有，则表示请求超时
		if(respnMessage == null) {
			if(timeOutMessages.get(msgID) == null) {
				connectManager.popTask(sessionId);
				log.error("被客户端请求走，并且超时.");
				throw new BizException(MessageCode.E10013);
			}
			//这时消息可能还没有返回,继续等待
			log.warn("等待返回结果");
			EncryptUtils.doWait(tom);
			respnMessage = respnMessages.remove(msgID);
			if(respnMessage == null) {
				connectManager.popTask(sessionId);
				log.error("被客户端请求走，等待后超时.");
				throw new BizException(MessageCode.E10013);
			}
			connectManager.popTask(sessionId);
			return respnMessage;
			
		}
		connectManager.popTask(sessionId);
		return respnMessage;
		
	}
	
	public List<MessageHolder> sendHeartbeatRequestAndWait(String encryptorId, long timeout, short commandid) {
		if(connectManager.hasClient()) {
			throw new BizException(MessageCode.E10014);
		}
		
		List<MessageHolder>  holders = new ArrayList<MessageHolder>();
		//指定单台加密机情况
		if(!Constants.ALL_ENCRYTORS.equals(encryptorId)) {
			IoSession session = connectManager.getSessionByEncryptorId(encryptorId);
			if(session == null) {
				throw new BizException(MessageCode.E10015);
			}
			holders.add(getHeartbeatResponse(session, timeout, commandid));
			return holders;
		}
		//所有加密机情况
		List<IoSession> sessions = connectManager.getAllSessions();
		for(IoSession session : sessions) {
			if(session == null) {
				throw new BizException(MessageCode.E10015);
			}
			holders.add(getHeartbeatResponse(session, timeout, commandid));
		}
		return holders;
	}
	
	public MessageHolder getHeartbeatResponse(IoSession session, long timeout, short commandid) {
		short sendCommand = MsgConstants.SERVER_REQ_HEARTBEAT;
		MessageHolder requestHolder = MessageUtils.getHeartbeatRequestForStatus();
		// 设置消息ID
		long msgID = MessageSequenceGenerator.getNextMessageId();
		requestHolder.setId(msgID);
		requestHolder.setSequence(msgID);
		HeartbeatRequest er = (HeartbeatRequest)requestHolder.getValue();
		er.setSequence(msgID);

		log.info("sequence:" + msgID);
		// 放入队列
		TimeoutMessage tom = new TimeoutMessage(requestHolder, timeout,sendCommand, commandid);
		sentMessages.put(msgID, tom);
		session.write(requestHolder);
		log.info("sessionId:" + session.getId() + ",address:" +  session.getRemoteAddress());
		// 等待消息的返回
		EncryptUtils.doWait(tom);
		// 从返回队列中获取返回值
		MessageHolder respnMessage = respnMessages.remove(msgID);
		if (respnMessage != null) {
			respnMessage.setSessionId(session.getId());
			return respnMessage;
		} else {
			log.error("response time out");
			throw new BizException(MessageCode.E10013);
		}
	}
	
	/**
	 * 消息到达，进行通知
	 * 
	 * @param message
	 */
	public void notifyBack(MessageHolder respnMessage) {
		Long seqID = respnMessage.getSequence();
		if (seqID == 0) {
			log.warn("Response message sequence is 0! Message: {}", respnMessage);
		}
		log.info("seqID:" + seqID);
		log.info("sentMessages:" + sentMessages);
		// pop出来
		TimeoutMessage tom = sentMessages.get(seqID);
		if (tom != null) {
			// 发出通知 唤醒
			if (tom.getWaitCommand() > 0 && tom.getWaitCommand() == respnMessage.getCommand()) {
				// 把响应消息放入队列
				respnMessages.put(seqID, respnMessage);
				
				// 通知其它线程去取响应消息
				EncryptUtils.doNotifyAll(tom);
				log.info("notify:" + tom);
				//移除
				sentMessages.remove( seqID );
			} else{
				//对一个请求会有多个消息响应的处理，还没有等待到最后一个命令，还需要等待。
				log.warn("no notify object in sentMessages.");
			}
		}
	}
	
	public TimeoutMessage getTask() {
		for(Entry<Long, TimeoutMessage> entry : sentMessages.entrySet()) {
			//心跳包跳过，只返回请求加密包
			TimeoutMessage tm = entry.getValue();
			if(tm.getWaitCommand() == MsgConstants.CLIENT_RES_HEARTBEAT) {
				continue;
			}
			if(tm.getStatus() == 0) {
				return entry.getValue();
			}
		}
		return null;
	}
	
	// 用于返回消息扫描
	public static class TimeoutMessage {
		private Date sendTime = new Date();

		// 等待client返回10分钟
		private long timeout;

		private short sendCommand;
		
		// 等待的消息类型，响应的消息会有多个，命令不一样
		private short waitCommand;

		MessageHolder message;
		
		//0:可以发送，1：已经在处理，不能发送此任务
		private int status;

		public TimeoutMessage(MessageHolder message, long timeout,short sendCommand,
				short waitCommand) {
			this.message = message;
			this.timeout = timeout;
			this.sendCommand = sendCommand;
			this.waitCommand = waitCommand;
		}

		public boolean isTimeOut() {
			Date now = new Date();
			if (now.getTime() - sendTime.getTime() > timeout) {
				return true;
			} else {
				return false;
			}
		}

		public long getTimeout() {
			return timeout;
		}

		public void setTimeout(long timeout) {
			this.timeout = timeout;
		}

		public short getSendCommand() {
			return sendCommand;
		}

		public void setSendCommand(short sendCommand) {
			this.sendCommand = sendCommand;
		}

		public short getWaitCommand() {
			return waitCommand;
		}

		public void setWaitCommand(short waitCommand) {
			this.waitCommand = waitCommand;
		}

		public MessageHolder getMessage() {
			return message;
		}

		public void setMessage(MessageHolder message) {
			this.message = message;
		}
		
		public String toString() {
			return message.toString();
		}

		public Date getSendTime() {
			return sendTime;
		}

		public void setSendTime(Date sendTime) {
			this.sendTime = sendTime;
		}

		public int getStatus() {
			return status;
		}

		public void setStatus(int status) {
			this.status = status;
		}

	}
	
	class CheckTimeOutThread extends Thread {
		public CheckTimeOutThread() {
			setName("Check TimeOut Thread");
			this.setDaemon(true);
		}

		public void run() {
			log.info("check response time out thread run...... ");
			while (true) {
				try {
					for (Long key : sentMessages.keySet()) {
						TimeoutMessage timeoutMessage = sentMessages.get(key);
						if (timeoutMessage.isTimeOut()) {
							TimeoutMessage tom = sentMessages.remove(key);
							//可能这个时候，消息刚好返回了，sentMessages.remove(key)会返回null
							if( tom!=null ){
								timeOutMessages.put(key, tom);
								// 通知其它线程去取响应消息
								EncryptUtils.doNotifyAll(tom);
								log.warn("get response timeout:{}. message:{}", tom.timeout, tom.message);
							}
						}
					}
					log.debug("check complete.it will check 2 seconds later.");
					Thread.sleep(2000);
				} catch (Exception e) {
					log.error("Check TimeOut Thread exception.", e);
				}
			}
		}
	}
}
